package com.atguigu1.core.acc

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 * @description: 缓存案例
 * @time: 2021-03-12 11:45
 * @author: baojinlong
 **/
object Spark01Acc {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("persistDemo")
    // 设置rdd分区数字
    val sparkContext = new SparkContext(conf)
    val rdd: RDD[Int] = sparkContext.makeRDD(Seq(1, 2, 3, 4))
    // reduce:分区内计算,分区间计算
    val result: Int = rdd.reduce(_ + _)
    println(result)
    var sum: Int = 0
    // sum的值在分布式计算中的值是不确定的.
    rdd.foreach(item => {
      sum += item
    })
    // sum在计算完成后并没有从Executor中返回来到Driver[累加器功能:把Executor端变量信息聚合到Driver端,在Driver程序中定义的变量,在Executor端的每个Task都会得到这个变量的一份新
    //的副本,每个task更新这些副本的值后,传回Driver端进行merge]
    println(sum)
    // 获取系统累加器spark默认提供了简单数据聚合的累加器
    val sumAccumulator: LongAccumulator = sparkContext.longAccumulator("sum")
    rdd.foreach(item => {
      // 使用累加器
      sumAccumulator.add(item)
    })
    println("finalResult " + sumAccumulator.value)

    // 注意:下面转换算子中调用累加器,如果没有行动算子的话那么不会执行
    val mapRdd: RDD[Int] = rdd.map(item => {
      // 使用累加器
      sumAccumulator.add(item)
      item
    })
    // 多加:由于累加器是全局的,所以只要是行动算子那么就会再加
    mapRdd.collect
    mapRdd.collect
    sparkContext.stop
  }

}